// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/feed/archive/system.h"

#include <memory>

#include "coin2/exchange/base/executor/archive_executor.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/exchange/base/feed/ws/feed_subscription_spec.h"
#include "coin2/flow/topic_subscription.h"

namespace coin2::feed::archive::impl {

RawArchiveFeedSystem::RawArchiveFeedSystem(
    const std::string& system_name,
    const MarketExchangeApi& mea,
    const coin2::exchange::di::ConstantInjection& ci,
    const FeedSystemConfig& config,
    ArchiveSyncExecutor* archive_executor,
    exchange::base::feed::IFeedSubscriber* callback)
    : IFeedSubsystem(system_name, mea),
      FeedArchiveMixin(archive_executor),
      config_(config) {
  AddCallback(callback);

  const auto& archive = config_.archive();
  ValidateArchiveConfig(archive);
  auto interval = FromIntervalConfig(archive.interval());
  auto warmup = DurationFromString(archive.read_warmup_duration());

  auto products = ProductRequestsFromConfig(mea_.String(), config.products(), interval.first);

  if (config.archive_recipes_size() > 0) {
    for (const auto& recipe : config.archive_recipes()) {
      auto raw_parser = BuildRawParserConfigForRecipe(recipe, mea, ci);
      auto processor = exchange::factory::feed::NewRawProcessor(
          ci, system_name, mea, raw_parser, recipe);
      processor->mutable_upd_mgr()->SetRecipe(recipe);
      processors_.push_back(std::make_pair(recipe, std::move(processor)));
    }
  } else {
    auto processor = exchange::factory::feed::NewRawProcessor(
        ci, system_name, mea, config.raw_parser(), config.recipe());
    processors_.push_back(std::pair{config.recipe(), std::move(processor)});
  }

  for (auto& prod : products) {
    if (callback != nullptr) {
      for (auto& [recipe, processor] : processors_) {
        processor->AddProduct(*prod, callback);
      }
    }
  }

  std::set<std::string> topics;
  if (config_.archive().arbitration()) {
    int num_workers = config_.archive().num_workers();
    for (int worker_id = 1; worker_id <= num_workers; ++worker_id) {
      for (auto& [recipe, processor] : processors_) {
        auto worker_topics = topic_register_.RegisterFlowTopicsArb(
          mea,
          std::to_string(worker_id),
          processor->upd_mgr().products(),
          processor.get(),
          archive_executor_->distributor(),
          ci.GetArbitrationConfigs(),
          recipe);
        topics.insert(worker_topics.begin(), worker_topics.end());
      }
    }
  } else if (config_.archive().shift()) {
    // worker 1/2 topics use the same processor
    for (auto& [recipe, processor] : processors_) {
      int num_workers = config_.archive().num_workers();
      for (int worker_id = 1; worker_id <= num_workers; ++worker_id) {
        auto worker_topics = topic_register_.RegisterFlowTopics(
          mea,
          std::to_string(worker_id),
          processor->upd_mgr().products(),
          processor.get(),
          archive_executor_->distributor());
        topics.insert(worker_topics.begin(), worker_topics.end());
      }
    }
  } else {
    for (auto& [recipe, processor] : processors_) {
      topics = topic_register_.RegisterFlowTopics(
          mea,
          std::to_string(archive.worker()),
          processor->upd_mgr().products(),
          processor.get(),
          archive_executor_->distributor());
    }
  }

  LOG(INFO) << "[RawFeed System]";
  LOG(INFO) << "              mea: " << mea_.String();
  LOG(INFO) << "interval - warmup: " << DebugStringFromDuration(warmup);
  LOG(INFO) << "interval -  start: " << Iso8601FromTimestamp(interval.first);
  LOG(INFO) << "interval -    end: " << Iso8601FromTimestamp(interval.second);

  reader_ = std::make_unique<flow::FlowReader>(
      archive.machine(),
      archive.flow_root_dir(),
      std::move(topics),
      interval.first - warmup,
      interval.second,
      archive.max_record_size(),
      archive.shift());
  interval_first_ = interval.first - warmup;
}

int64_t RawArchiveFeedSystem::PeekTimestamp() {
  if (!has_packet_) {
    while (reader_->Read(&packet_)) {
      if (packet_.timestamp < interval_first_) {
        continue;
      }
      has_packet_ = true;
      packet_.timestamp += config_.archive().offset_ns();
      break;
    }
    if (!has_packet_) {
      return 0;
    }
  }
  return packet_.timestamp;
}

void RawArchiveFeedSystem::ConsumeRecord() {
  CHECK_THROW(has_packet_) << "Cannot consume a packet";
  has_packet_ = false;
  archive_executor_->distributor()->OnTopicRecord(packet_.topic_name, &packet_);
}

RawParserConfig RawArchiveFeedSystem::BuildRawParserConfigForRecipe(
    const std::string& recipe,
    const MarketExchangeApi& mea,
    const coin2::exchange::di::ConstantInjection& ci) {
  const auto& consts = ci.GetFeedSubscriptionSpec(mea);
  RawParserConfig raw_parser;
  raw_parser.CopyFrom(config_.raw_parser());
  raw_parser.clear_channels();
  for (const auto& ch : consts.recipes.at(recipe)) {
    raw_parser.add_channels(ch);
  }
  // light_bbo override
  if (recipe.find("light_bbo") != std::string::npos) {
    exchange::feed::RawParserConfig light_bbo_override;
    light_bbo_override.set_enable_diff_book(false);
    light_bbo_override.set_publish_bbo_book_only(true);
    light_bbo_override.set_book_level(5);
    raw_parser.MergeFrom(light_bbo_override);
    CHECK_EQ(recipe.find("move2bp"), std::string::npos);
  } else if (recipe.find("move2bp") != std::string::npos) {
    exchange::feed::RawParserConfig move2bp_override;
    move2bp_override.set_sampling_frequency_bps(2);
    move2bp_override.set_book_level(5);
    if (recipe.find("trade_move2bp") != std::string::npos) {
      move2bp_override.set_trade_sampling_frequency_bps(2);
    }
    raw_parser.MergeFrom(move2bp_override);
  }
  return raw_parser;
}
}  // namespace coin2::feed::archive::impl
